package com.scala.learn.project

import java.net.URL

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

import scala.collection.mutable


/**
  * 计算每个学科的前3名
  * 类似于二次排序
  */
object FavTeacher {


  def main(args: Array[String]): Unit = {
    //初始化conf
    var conf = new SparkConf().setAppName("FavTeacher")

    var sc = new SparkContext(conf)
    //加载文件
    val rdd1 = sc.textFile("")
    // https://ke.qq.com/bigData/xiaowang/index.html
    val subjectAndTeacher = rdd1.map(line => {
      var url = new URL(line)
      var path = url.getPath
      var res = path.split("/")
      //bigData
      var subject = res(1)
      var teacher = res(2)
      (subject, teacher)
    })


    //1、将同样学科，同样老师的聚合在一起；
    val reduced: RDD[((String, String), Int)] = subjectAndTeacher.map((_, 1)).reduceByKey(_ + _)

    //使用缓存
    reduced.cache()
    //2.1将聚合过得数据中学科的聚合到一起
    val subjects: Array[String] = reduced.map(_._1._1).distinct().collect()

    //2.2获取分区器
    var subPartitioner = new SubPartitioner(subjects)

    //2、使用自定义分区，将同一个学科的放在同一个分区，分区的数量怎么定？


    val partitioned: RDD[(String, (String, Int))] = reduced.map(t => (t._1._1, (t._1._2, t._2))).partitionBy(subPartitioner)

    // val partitioned: RDD[((String, String), Int)] = reduced.partitionBy(subPartitioner)
    //拿出整个分区，在分区中对集合进行排序，取出2个，并返回iterator

    val result: RDD[(String, (String, Int))] =
      partitioned.mapPartitions(iter => iter.toList.sortBy(_._2._2).reverse.take(2).iterator)

    result.saveAsTextFile("")

    sc.stop()

  }

  /**
    * 自定义分区器
    * 实现package org.apache.spark.Partitioner类
    */


  class SubPartitioner(subjects: Array[String]) extends Partitioner {
    val rules = new mutable.HashMap[String, Int]
    var i = 1
    for (sub <- subjects) {
      rules += (sub -> i)
    }

    //分区的数量
    override def numPartitions: Int = subjects.length

    //分区的规则,什么时候调用，传过来的值是什么，
    //传过来的是根据
    // val partitioned: RDD[((String, String), Int)] = reduced.partitionBy(subPartitioner)
    //所以key是一个
    override def getPartition(key: Any): Int = {
      rules.getOrElse(key.toString, 0)
    }
  }

}
